package com.ymt.reliable.message.service.impl;

import com.ymt.reliable.message.dao.TransactionMessageMapper;
import com.ymt.reliable.message.domain.TransactionMessage;
import com.ymt.reliable.message.domain.TransactionMessageFactory;
import com.ymt.reliable.message.enums.MessageStatus;
import com.ymt.reliable.message.exception.TransactionMessageNotExistException;
import com.ymt.reliable.message.queue.service.MessageQueueService;
import com.ymt.reliable.message.service.TransactionMessageIdService;
import com.ymt.reliable.message.service.TransactionMessageService;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.stereotype.Service;
import org.springframework.transaction.annotation.Transactional;

import java.util.Date;
import java.util.HashMap;
import java.util.List;
import java.util.Map;

@Service(value = "transactionMessageService")
@Transactional
public class TransactionMessageServiceImpl implements TransactionMessageService {
    private static final Logger LOGGER = LoggerFactory.getLogger(TransactionMessageServiceImpl.class);

    @Value("${retry.maxsendtimes:6}")
    private Integer maxSendTimes;

    @Autowired
    private MessageQueueService messageQueueService;
    @Autowired
    private TransactionMessageMapper transactionMessageMapper;
    @Autowired
    private TransactionMessageIdService transactionMessageIdService;

    /**
     * 验证messageId是否正确
     *
     * @param messageId
     */
    private TransactionMessage messageExist(String messageId) throws TransactionMessageNotExistException {
        TransactionMessage message = transactionMessageMapper.get(messageId);
        if (message == null) {
            throw new TransactionMessageNotExistException(String.format("消息编号%s不存在", messageId));
        }
        return message;
    }

    /**
     * 重发消息
     *
     * @param message
     */
    private void reSendMessage(TransactionMessage message) {
        TransactionMessage transactionMessage = new TransactionMessage();
        transactionMessage.setId(message.getId());
        transactionMessage.setMessageSendTimes(message.getMessageSendTimes());
        transactionMessage.addMessageSendTimes();

        //验证重试次数是否已经大于等于最大重试次数
        if (transactionMessage.getMessageSendTimes() >= maxSendTimes) {
            transactionMessage.setAreadlyDead("1");
            LOGGER.info(">>> 消息超出最大重试次数标识死亡!消息编号:{},重试次数:{}", message.getMessageId(), message.getMessageSendTimes());
        }
        transactionMessageMapper.update(transactionMessage);

        messageQueueService.sendMessageToMessageQueue(message.getConsumerQueue(), message.getMessageId(), message.getMessageBody());
    }

    @Override
    public String savePreparMessage(String messageBody, String consumerQueue, String field1, String field2, String field3) {
        TransactionMessage transactionMessage = TransactionMessageFactory.createTransactionMessage(messageBody, consumerQueue, field1, field2, field3);
        transactionMessage.setMessageId(transactionMessageIdService.generateId());
        transactionMessageMapper.insert(transactionMessage);

        LOGGER.info(">>> 存储预发布消息! 消息编号:{},队列名称:{}", transactionMessage.getMessageId(), consumerQueue);
        return transactionMessage.getMessageId();
    }

    @Override
    public synchronized void confirmAndSend(String messageId) throws TransactionMessageNotExistException {
        Date sysTime = new Date();
        TransactionMessage message = messageExist(messageId);
        if (!MessageStatus.WAITING_CONFIRM.getFlag().equals(message.getStatus())) {
            return;
        }
        message.setStatus(MessageStatus.SENDING.getFlag());
        message.setLastPushTime(sysTime);
        transactionMessageMapper.update(message);

        messageQueueService.sendMessageToMessageQueue(message.getConsumerQueue(), message.getMessageId(), message.getMessageBody());
        LOGGER.info(">>> 确认消息并且发送到消息中间件! 消息编号:{},队列名称:{}", messageId, message.getConsumerQueue());
    }

    @Override
    public void directSendMessage(String messageBody, String consumerQueue) {
        messageQueueService.sendMessageToMessageQueue(consumerQueue, "", messageBody);
        LOGGER.info(">>> 直接发送到消息中间件! 队列名称:{}", consumerQueue);
    }

    @Override
    public void retry(String messageId) throws TransactionMessageNotExistException {
        TransactionMessage message = messageExist(messageId);

        reSendMessage(message);

        LOGGER.info(">>> 重新发送消息! 消息编号:{}", messageId);
    }

    @Override
    public void delete(String messageId) {
        transactionMessageMapper.deleteByMessageId(messageId);
        LOGGER.info(">>> 删除消息! 消息编号:{}", messageId);
    }

    @Override
    public List<TransactionMessage> queryRetryList() {
        Map<String, Object> params = new HashMap<>();
        return transactionMessageMapper.queryRetryList(params);
    }

    @Override
    public List<TransactionMessage> queryRetryList(Long timeout) {
        Map<String, Object> params = new HashMap<>();
        params.put("timeout", timeout);
        return transactionMessageMapper.queryRetryList(params);
    }

    @Override
    public List<TransactionMessage> queryConfirmList(Long timeout, Integer maxconfirmtimes) {
        Map<String, Object> params = new HashMap<>();
        params.put("timeout", timeout);
        params.put("maxconfirmtimes", maxconfirmtimes);
        return transactionMessageMapper.queryConfirmList(params);
    }

    @Override
    public Integer update(TransactionMessage message) {
        return transactionMessageMapper.update(message);
    }
}
